package org.example.rocketmqstudy.consumer;

import io.netty.channel.DefaultChannelId;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.*;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.List;
import java.util.concurrent.atomic.AtomicLong;

public class OrderlyConsumer {
    public static void main(String[] args) throws MQClientException {
        DefaultChannelId.newInstance(); // 解决 RemotingTimeoutException: invokeSync call timeout 异常

        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("orderly-consumer");
        consumer.setNamesrvAddr("8.138.36.56:9876");
        consumer.subscribe("test-order", "*");

        //注册回调接口来处理从Broker中收到的消息
        consumer.registerMessageListener(new MessageListenerOrderly() {
            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
                String now = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date());
                System.out.printf("%s [%s] 收到消息: %s %n", now, Thread.currentThread().getName(), new String(msgs.get(0).getBody()));
                return ConsumeOrderlyStatus.SUCCESS;
            }
        });
        // 启动Consumer
        consumer.start();
        System.out.printf("Consumer Started.%n");
    }
}
